package com.shujia.MR;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.BinaryComparable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.List;

/**
 * 读取Hbase中的Student数据，并且对学生班级和年龄进行统计，再将统计结果写入至HBASE中
 * 并在要求该学生的 ID大于 15001001
 */
public class Hbase2Hbase {
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {


        Configuration conf = new Configuration();
        conf.set("hbase.zookeeper.quorum","node1,node2,master");

        Job job = Job.getInstance(conf);
        job.setJarByClass(Hbase2Hbase.class);
        job.setJobName("Hbase2Hbase");


        // 设置Mapper
        /**
         * public static void initTableMapperJob(TableName table,
         *       Scan scan,
         *       Class<? extends TableMapper> mapper,
         *       Class<?> outputKeyClass,
         *       Class<?> outputValueClass,
         *       Job job)
         */

        // 要求该学生的 ID大于 15001001 可以使用RowFilter

        Scan scan = new Scan();
        RowFilter rowFilter = new RowFilter(
                CompareFilter.CompareOp.GREATER
                , new BinaryComparator(Bytes.toBytes("15001001!"))
        );

        scan.setFilter(rowFilter);


        TableMapReduceUtil.initTableMapperJob(
                TableName.valueOf("api:tb1")
                ,scan
                ,MyHbaseMapper.class
                ,Text.class
                ,IntWritable.class
                ,job
        );

        // 设置Reducer
        TableMapReduceUtil.initTableReducerJob(
            "api:countRes"
                ,MyHbaseReducer.class
                ,job
        );

        // 提交
        job.waitForCompletion(true);
    }

    // Mapper

    /**
     * 由于Mapper端需要取出学生的班级及年龄，所以对应输出的Key和Value是 Text, IntWritable
     */
    public static class MyHbaseMapper extends TableMapper<Text, IntWritable> {
        @Override
        protected void map(ImmutableBytesWritable key, Result value, Mapper<ImmutableBytesWritable, Result, Text, IntWritable>.Context context) throws IOException, InterruptedException {
            List<Cell> cells = value.listCells();
            int age = 0;
            String clazz = null;
            for (Cell cell : cells) {
                // 通过CellUtil工具类取出其列名称和对应的值
                //  String clazz = Bytes.toString(value.getValue(Bytes.toBytes("info2"), Bytes.toBytes("clazz")));
                //   以上代码和CellUtil相比，代码简洁，但是容易发生空指针异常
                if (Bytes.toString(CellUtil.cloneQualifier(cell)).equals("age")) {
                    age = Bytes.toInt(CellUtil.cloneValue(cell));
                }
                if (Bytes.toString(CellUtil.cloneQualifier(cell)).equals("clazz")) {
                    clazz = Bytes.toString(CellUtil.cloneValue(cell));
                }
            }
            context.write(new Text(clazz + "|" + age), new IntWritable(1));
        }
    }


    // Reducer

    /**
     * 由于数据是需要写入到HBASE中 --》需要继承TableReducer
     * 通过查看TableReducer类发现其有 三个泛型 TableReducer<KEYIN, VALUEIN, KEYOUT>
     * 并且其继承父类时对应ValueOut的泛型为Mutation类，需要查看Mutation类
     * --》 该类为抽象类 无法直接使用，所以需要查看其子实现类 发现有Put类
     * --》 那么由此可以知道 后续的结果数据，可以直接包装成一个Put去写入到HBASE中
     * --》 由于Put中已经可以设置RowKey 那么对于KEYOUT 那么可以设置成 NullWritable
     */
    public static class MyHbaseReducer extends TableReducer<Text, IntWritable, NullWritable> {
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, NullWritable, Mutation>.Context context) throws IOException, InterruptedException {
            int num = 0;
            for (IntWritable value : values) {
                num += value.get();
            }

            Put put = new Put(Bytes.toBytes(key.toString()));

            /**
             * public KeyValue(final byte [] row, final byte [] family,
             *       final byte [] qualifier, final byte [] value)
             */
            KeyValue keyValue = new KeyValue(
                    Bytes.toBytes(key.toString()),
                    Bytes.toBytes("info1"),
                    Bytes.toBytes("num"),
                    Bytes.toBytes(num)
            );

            put.add(keyValue);

            // 写出结果
            context.write(NullWritable.get(),put);
        }
    }


}
